package com.atguigu.process;

import com.atguigu.bean.WaterSensor;
import com.atguigu.functions.MapFunction2Impl;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.TimerService;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

import java.time.Duration;

/**
 * @author gmd
 * @desc 按键盘分区处理函数（KeyedProcessFunction）的使用+定时器
 * @since 2024-11-30 10:09:11
 */
public class KeyedProcessTimer {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        SingleOutputStreamOperator<WaterSensor> sensorDS = env
                .socketTextStream("127.0.0.1", 7777)
                .map(new MapFunction2Impl())
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy
                                .<WaterSensor>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                                .withTimestampAssigner((element, ts) -> element.getTs() * 1000L)
                );
        KeyedStream<WaterSensor, String> sensorKS = sensorDS.keyBy(sensor -> sensor.getId());


        SingleOutputStreamOperator<String> process = sensorKS.process(
                new KeyedProcessFunction<String, WaterSensor, String>() {
                    /**
                     * 每当接收到一条数据时调用此方法
                     *
                     * @param value   当前接收到的数据
                     * @param ctx     上下文对象，提供了对定时器和状态的访问
                     * @param out     输出收集器，用于输出处理结果
                     */
                    @Override
                    public void processElement(WaterSensor value, Context ctx, Collector<String> out) throws Exception {
                        // 获取当前数据的key
                        String currentKey = ctx.getCurrentKey();

                        // 定时器注册
                        TimerService timerService = ctx.timerService();

                        // 1、事件时间的案例
                        Long currentEventTime = ctx.timestamp(); // 数据中提取出来的事件时间
                        timerService.registerEventTimeTimer(5000L);
                        System.out.println("当前key=" + currentKey + "，当前时间=" + currentEventTime + "，注册了一个5s的定时器");

                        // 2、处理时间的案例
//                        long currentTs = timerService.currentProcessingTime();
//                        timerService.registerProcessingTimeTimer(currentTs + 5000L);
//                        System.out.println("当前key=" + currentKey + ",当前时间=" + currentTs + ",注册了一个5s后的定时器");


                        // 3、获取 process的 当前watermark
//                        long currentWatermark = timerService.currentWatermark();
//                        System.out.println("当前数据=" + value + ",当前watermark=" + currentWatermark);



                        // 注册定时器： 处理时间、事件时间
//                        timerService.registerProcessingTimeTimer();
//                        timerService.registerEventTimeTimer();
                        // 删除定时器： 处理时间、事件时间
//                        timerService.deleteEventTimeTimer();
//                        timerService.deleteProcessingTimeTimer();

                        // 获取当前时间进展： 处理时间-当前系统时间，  事件时间-当前watermark
//                        long currentTs = timerService.currentProcessingTime();
//                        long wm = timerService.currentWatermark();
                    }


                    /**
                     * 当时间进展到定时器注册的时间时，调用该方法
                     *
                     * @param timestamp 当前时间进展，就是定时器被触发时的时间
                     * @param ctx       上下文对象，提供了对定时器和状态的访问
                     * @param out       采集器，用于输出处理结果
                     */
                    @Override
                    public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {
                        super.onTimer(timestamp, ctx, out);
                        String currentKey = ctx.getCurrentKey();

                        System.out.println("key=" + currentKey + "，现在时间是：" + timestamp + "定时器触发");
                    }
                }
        );

        process.print();
        env.execute();
    }

    /**
     * 定时器（Timer）使用说明
     * =====================================
     * 1. 定时器仅在 Keyed Process Function 中可用
     * 2. 事件时间定时器（Event Time Timer）通过 watermark 触发
     *    watermark 需要大于或等于注册的时间
     *    注意：watermark 的计算公式为：当前最大事件时间 - 等待时间 - 1ms
     *          由于减去 1ms，因此会推迟一条数据的触发
     *          例如：5s 的定时器，如果等待时间为 3s，则 watermark 需要达到 9s - 3s - 1ms = 5999ms 才能触发
     * 3. 在 Process Function 中获取当前 watermark，显示的是上一次的 watermark
     *    这是因为 Process Function 还没有接收到这条数据对应生成的新 watermark
     */
}
